package com.ys.http.body;


import com.ys.http.download.OnDownloadListener;

import java.io.IOException;
import java.util.concurrent.TimeUnit;

import io.reactivex.rxjava3.android.schedulers.AndroidSchedulers;
import io.reactivex.rxjava3.core.Observable;
import io.reactivex.rxjava3.disposables.Disposable;
import io.reactivex.rxjava3.schedulers.Schedulers;
import okhttp3.MediaType;
import okhttp3.ResponseBody;
import okio.Buffer;
import okio.BufferedSource;
import okio.ForwardingSource;
import okio.Okio;
import okio.Source;

/**
 * 下载响应体，用来处理下载进度
 */
public class DownResponseBody extends ResponseBody {

    private ResponseBody mResponseBody;
    private final OnDownloadListener mDownloadListener;
    // BufferedSource 是okio库中的输入流，这里就当作inputStream来使用。
    private BufferedSource mBufferedSource;
    private Disposable mProgressDisposable;
    private long mCurrentSize; //当前已读写的字节数
    private long mSecondSize; //每秒读取的字节数
    private long mTotalSize;

    public DownResponseBody(OnDownloadListener listener) {
        this.mDownloadListener = listener;
    }

    public void setResponseBody(ResponseBody body) {
        this.mResponseBody = body;
    }

    private void initDownloadProgress() {
        mTotalSize = contentLength();
        mProgressDisposable = Observable.interval(1, TimeUnit.SECONDS)
                .subscribeOn(Schedulers.io())
                .observeOn(AndroidSchedulers.mainThread())
                .subscribe(val -> {
                    long size = mCurrentSize - mSecondSize;
                    mSecondSize += size;
                    int progress = 0;
                    if (mTotalSize > 0) {
                        progress = (int) (mCurrentSize * 100 / mTotalSize);
                    }
                    mDownloadListener.onProgress(progress, mCurrentSize, size);
                    if (mCurrentSize == mTotalSize) {
                        mProgressDisposable.dispose();
                    }
                });
    }

    @Override
    public MediaType contentType() {
        return mResponseBody.contentType();
    }

    @Override
    public long contentLength() {
        return mResponseBody.contentLength();
    }

    @Override
    public BufferedSource source() {
        if (mBufferedSource == null) {
            initDownloadProgress();
            mBufferedSource = Okio.buffer(source(mResponseBody.source()));
        }
        return mBufferedSource;
    }

    private Source source(Source source) {
        return new ForwardingSource(source) {
            @Override
            public long read(Buffer sink, long byteCount) throws IOException {
                long bytesRead = super.read(sink, byteCount);
                if (bytesRead != -1) {
                    mCurrentSize += bytesRead;
                }
                return bytesRead;
            }
        };
    }

    @Override
    public void close() {
        super.close();
        dispose();
    }

    public void dispose() {
        if (mProgressDisposable != null) {
            mProgressDisposable.dispose();
            mProgressDisposable = null;
        }
    }
}
